package com.lazar.ods

import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.lazar.model.{AreaInfo, DataInfo, OrderProduct, Payment, ProductCategory, ProductInfo, Shop, ShopOrg, TableObject}
import com.lazar.utils.ConnHBase
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.conf.Configuration
import org.apache.flink.configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Delete, Put, Table}

class SinkHBase extends RichSinkFunction[util.ArrayList[TableObject]]{
  var connection : Connection = _
  var hbTable : Table = _
  /**
   * 实例化 HBase
   * connection
   * hbTable
   * @param parameters
   */
  override def open(parameters: configuration.Configuration): Unit = {
    connection = new ConnHBase().connToHbase
    hbTable = connection.getTable(TableName.valueOf("lagou_area"))
  }

  override def close(): Unit = {
    if(hbTable != null) {
      hbTable.close()
    }
    if (connection != null) {
      connection.close()
    }
  }

  /**
   * 每来一条数据，会执行一次
   * @param value
   * @param context
   */
  override def invoke(value: util.ArrayList[TableObject], context: SinkFunction.Context[_]): Unit = {
    value.forEach(x => {
      println(x.toString)
      val database: String = x.database
      val tableName: String = x.tableName
      val typeInfo: String = x.typeInfo

      // area
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_area")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            insertArea(hbTable,info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            insertArea(hbTable,info)
          })
        } else if (typeInfo.equalsIgnoreCase("delete")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            deleteArea(hbTable,info)
          })

        }
      }

      // order product
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_order_product")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: OrderProduct = JSON.parseObject(x.dataInfo, classOf[OrderProduct])
            insertOrderProduct(connection.getTable(TableName.valueOf("lagou_order_product")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }

      //payments
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_payments")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: Payment = JSON.parseObject(x.dataInfo, classOf[Payment])
            insertPayment(connection.getTable(TableName.valueOf("lagou_payments")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }

      //product category
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_product_category")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: ProductCategory = JSON.parseObject(x.dataInfo, classOf[ProductCategory])
            insertProductCategory(connection.getTable(TableName.valueOf("lagou_product_category")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }

      //product info
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_product_info")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: ProductInfo = JSON.parseObject(x.dataInfo, classOf[ProductInfo])
            insertProductInfo(connection.getTable(TableName.valueOf("lagou_product_info")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }
      //product info
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_shop_admin_org")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: ShopOrg = JSON.parseObject(x.dataInfo, classOf[ShopOrg])
            insertShopOrg(connection.getTable(TableName.valueOf("lagou_shop_admin_org")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }
      //product info
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_shops")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: Shop = JSON.parseObject(x.dataInfo, classOf[Shop])
            insertShop(connection.getTable(TableName.valueOf("lagou_shops")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }

      // order
      if(database.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_trade_orders")) {
        if(typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: DataInfo = JSON.parseObject(x.dataInfo, classOf[DataInfo])
            insertTradeOrders(connection.getTable(TableName.valueOf("lagou_trade_orders")),info)
          })
        } else if(typeInfo.equalsIgnoreCase("update")) {

        } else if (typeInfo.equalsIgnoreCase("delete")) {

        }
      }

    })
  }

  //lagou_area省份城市区域表，根据id删除数据
  def deleteArea(hbTable: Table, areaInfo: AreaInfo): Unit = {
    val delete = new Delete(areaInfo.id.getBytes)
    hbTable.delete(delete)
  }

  def insertArea(hbTable: Table, areaInfo: AreaInfo) : Unit = {
   //println(areaInfo.toString)
    val put = new Put(areaInfo.id.getBytes())
    put.addColumn("f1".getBytes(), "name".getBytes(), areaInfo.name.getBytes())
    put.addColumn("f1".getBytes(), "pid".getBytes(), areaInfo.pid.getBytes())
    put.addColumn("f1".getBytes(), "sname".getBytes(), areaInfo.sname.getBytes())
    put.addColumn("f1".getBytes(), "level".getBytes(), areaInfo.level.getBytes())
    put.addColumn("f1".getBytes(), "citycode".getBytes(), areaInfo.citycode.getBytes())
    put.addColumn("f1".getBytes(), "yzcode".getBytes(), areaInfo.yzcode.getBytes())
    put.addColumn("f1".getBytes(), "mername".getBytes(), areaInfo.mername.getBytes())
    put.addColumn("f1".getBytes(), "lng".getBytes(), areaInfo.Lng.getBytes())
    put.addColumn("f1".getBytes(), "lat".getBytes(), areaInfo.Lat.getBytes())
    put.addColumn("f1".getBytes(), "pinyin".getBytes(), areaInfo.pinyin.getBytes())

    hbTable.put(put)
  }
  def insertOrderProduct(hbTable: Table,orderProduct: OrderProduct): Unit ={
    val put = new Put(orderProduct.id.getBytes())
    put.addColumn("f1".getBytes(), "orderId".getBytes(), orderProduct.orderId.getBytes())
    put.addColumn("f1".getBytes(), "productId".getBytes(), orderProduct.productId.getBytes())
    put.addColumn("f1".getBytes(), "productNum".getBytes(), orderProduct.productNum.getBytes())
    put.addColumn("f1".getBytes(), "productPrice".getBytes(), orderProduct.productPrice.getBytes())
    put.addColumn("f1".getBytes(), "money".getBytes(), orderProduct.money.getBytes())
    put.addColumn("f1".getBytes(), "extra".getBytes(), orderProduct.extra.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), orderProduct.createTime.getBytes())

    hbTable.put(put)
  }
  def insertPayment(hbTable: Table,payment: Payment): Unit ={
    val put = new Put(payment.id.getBytes())
    put.addColumn("f1".getBytes(), "payMethod".getBytes(), payment.payMethod.getBytes())
    put.addColumn("f1".getBytes(), "payName".getBytes(), payment.payName.getBytes())
    put.addColumn("f1".getBytes(), "payOrder".getBytes(), payment.payOrder.getBytes())
    put.addColumn("f1".getBytes(), "description".getBytes(), payment.description.getBytes())
    put.addColumn("f1".getBytes(), "online".getBytes(), payment.online.getBytes())

    hbTable.put(put)
  }
  def insertProductCategory(hbTable: Table,productCategory: ProductCategory): Unit ={
    val put = new Put(productCategory.catId.getBytes())
    put.addColumn("f1".getBytes(), "catName".getBytes(), productCategory.catName.getBytes())
    put.addColumn("f1".getBytes(), "parentId".getBytes(), productCategory.parentId.getBytes())
    put.addColumn("f1".getBytes(), "isShow".getBytes(), productCategory.isShow.getBytes())
    put.addColumn("f1".getBytes(), "sortNum".getBytes(), productCategory.sortNum.getBytes())
    put.addColumn("f1".getBytes(), "isDel".getBytes(), productCategory.isDel.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), productCategory.createTime.getBytes())
    put.addColumn("f1".getBytes(), "level".getBytes(), productCategory.level.getBytes())

    hbTable.put(put)
  }
  def insertProductInfo(hbTable: Table,productInfo: ProductInfo): Unit ={
    val put = new Put(productInfo.productId.getBytes())
    put.addColumn("f1".getBytes(), "productName".getBytes(), productInfo.productName.getBytes())
    put.addColumn("f1".getBytes(), "shopId".getBytes(), productInfo.shopId.getBytes())
    put.addColumn("f1".getBytes(), "price".getBytes(), productInfo.price.getBytes())
    put.addColumn("f1".getBytes(), "isSale".getBytes(), productInfo.isSale.getBytes())
    put.addColumn("f1".getBytes(), "status".getBytes(), productInfo.status.getBytes())
    put.addColumn("f1".getBytes(), "categoryId".getBytes(), productInfo.categoryId.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), productInfo.createTime.getBytes())
    put.addColumn("f1".getBytes(), "modifyTime".getBytes(), productInfo.modifyTime.getBytes())

    hbTable.put(put)
  }
  def insertShop(hbTable: Table,shop: Shop): Unit ={
    val put = new Put(shop.shopId.getBytes())
    put.addColumn("f1".getBytes(), "userId".getBytes(), shop.userId.getBytes())
    put.addColumn("f1".getBytes(), "shopId".getBytes(), shop.shopId.getBytes())
    put.addColumn("f1".getBytes(), "areaId".getBytes(), shop.areaId.getBytes())
    put.addColumn("f1".getBytes(), "shopName".getBytes(), shop.shopName.getBytes())
    put.addColumn("f1".getBytes(), "shopLevel".getBytes(), shop.shopLevel.getBytes())
    put.addColumn("f1".getBytes(), "status".getBytes(), shop.status.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), shop.createTime.getBytes())
    put.addColumn("f1".getBytes(), "modifyTime".getBytes(), shop.modifyTime.getBytes())

    hbTable.put(put)
  }
  def insertShopOrg(hbTable: Table,shopOrg: ShopOrg): Unit ={
    val put = new Put(shopOrg.id.getBytes())
    put.addColumn("f1".getBytes(), "parentId".getBytes(), shopOrg.parentId.getBytes())
    put.addColumn("f1".getBytes(), "orgName".getBytes(), shopOrg.orgName.getBytes())
    put.addColumn("f1".getBytes(), "orgLevel".getBytes(), shopOrg.orgLevel.getBytes())
    put.addColumn("f1".getBytes(), "isDelete".getBytes(), shopOrg.isDelete.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), shopOrg.createTime.getBytes())
    put.addColumn("f1".getBytes(), "updateTime".getBytes(), shopOrg.updateTime.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), shopOrg.createTime.getBytes())
    put.addColumn("f1".getBytes(), "isShow".getBytes(), shopOrg.isShow.getBytes())
    put.addColumn("f1".getBytes(), "orgType".getBytes(), shopOrg.orgType.getBytes())

    hbTable.put(put)
  }
  def insertTradeOrders(hbTable: Table, dataInfo: DataInfo): Unit = {
    val put = new Put(dataInfo.orderId.getBytes)
    put.addColumn("f1".getBytes,"modifiedTime".getBytes,dataInfo.modifiedTime.getBytes())
    put.addColumn("f1".getBytes,"orderNo".getBytes,dataInfo.orderNo.getBytes())
    put.addColumn("f1".getBytes,"isPay".getBytes,dataInfo.isPay.getBytes())
    put.addColumn("f1".getBytes,"orderId".getBytes,dataInfo.orderId.getBytes())
    put.addColumn("f1".getBytes,"tradeSrc".getBytes,dataInfo.tradeSrc.getBytes())
    put.addColumn("f1".getBytes,"payTime".getBytes,dataInfo.payTime.getBytes())
    put.addColumn("f1".getBytes,"productMoney".getBytes,dataInfo.productMoney.getBytes())
    put.addColumn("f1".getBytes,"totalMoney".getBytes,dataInfo.totalMoney.getBytes())
    put.addColumn("f1".getBytes,"dataFlag".getBytes,dataInfo.dataFlag.getBytes())
    put.addColumn("f1".getBytes,"userId".getBytes,dataInfo.userId.getBytes())
    put.addColumn("f1".getBytes,"areaId".getBytes,dataInfo.areaId.getBytes())
    put.addColumn("f1".getBytes,"createTime".getBytes,dataInfo.createTime.getBytes())
    put.addColumn("f1".getBytes,"payMethod".getBytes,dataInfo.payMethod.getBytes())
    put.addColumn("f1".getBytes,"isRefund".getBytes,dataInfo.isRefund.getBytes())
    put.addColumn("f1".getBytes,"tradeType".getBytes,dataInfo.tradeType.getBytes())
    put.addColumn("f1".getBytes,"status".getBytes,dataInfo.status.getBytes())

    hbTable.put(put)
  }
}
